package com.ruoyi.iot.handler;


import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.iot.config.MqttSubscribeService;
import com.ruoyi.iot.service.MqttMessageHandler;
import com.ruoyi.iotcommon.constant.RocketTopicConstant;
import com.ruoyi.iotcommon.domain.DeviceDataMessage;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/**
 * topic = /sys/${productKey}/${deviceKey}/thing/event/${channel}/post
 * topic = /sys/+/+/thing/event/+/post
 */

@Slf4j
@Service
@MqttSubscribeService(topic = "/sys/+/+/thing/event/+/post")
public class DeviceDataMqttMessageHandler implements MqttMessageHandler {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void process(String topic,String message) {
        try {
            String[] topicArray = topic.split("/");
            String productKey = topicArray[2];
            String deviceKey = topicArray[3];
            String channelKey = topicArray[6];
            JSONObject msgBody = JSONObject.parseObject(message);
            JSONObject paramsJson = msgBody.getJSONObject("params");
            // 将paramsJson转换为map
            Map<String, Object> params = new HashMap<>(paramsJson);

            //  发送rocketmq
            DeviceDataMessage deviceDataMessage = new DeviceDataMessage();
            deviceDataMessage.setProductKey(productKey);
            deviceDataMessage.setDeviceKey(deviceKey);
            deviceDataMessage.setChannelKey(channelKey);
            deviceDataMessage.setTime(System.currentTimeMillis());
            deviceDataMessage.setParams(params);
        
            //rocketMQTemplate.convertAndSend(RocketTopicConstant.DEVICE_DATA, deviceDataMessage);
            rocketMQTemplate.asyncSend(RocketTopicConstant.DEVICE_DATA, deviceDataMessage,new SendCallback(){

                @Override
                public void onException(Throwable arg0) {
                    log.error("设备数据转发失败",arg0);
                }

                @Override
                public void onSuccess(SendResult arg0) {
                    log.info("设备数据转发成功");
                }

            });
        } catch (Exception e) {
            log.error("解析设备数据异常",e);
        }
        
    }
}
